package com.dc.rxjava

import android.os.Bundle
import android.text.TextUtils
import android.util.Log
import android.widget.Button
import android.widget.EditText
import android.widget.TextView
import androidx.appcompat.app.AppCompatActivity
import com.jakewharton.rxbinding2.view.RxView
import com.jakewharton.rxbinding2.widget.RxTextView
import io.reactivex.*
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.Disposable
import io.reactivex.functions.Consumer
import io.reactivex.functions.Function
import io.reactivex.schedulers.Schedulers
import java.io.IOException
import java.util.concurrent.TimeUnit

/**
 * 1.功能防抖
 * 2.合并事件-联合判断
 * 3.数据源合并
 * 4.内存缓存、磁盘缓存获取数据
 * 5.联想搜索
 * 6.网络请求出错重连
 * 7.网络请求轮询
 */
class RxJavaActivity : AppCompatActivity() {

    private val TAG = "RxJava"

    private lateinit var name: EditText
    private lateinit var job: EditText
    private lateinit var age: EditText
    private lateinit var submit: Button
    private lateinit var search: EditText
    private lateinit var result: TextView

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_rxjava)
//        FunctionOperatorUtils.onExceptionResumeNext()

        name = findViewById(R.id.name)
        job = findViewById(R.id.job)
        age = findViewById(R.id.age)
        submit = findViewById(R.id.submit)
        search = findViewById(R.id.search)
        result = findViewById(R.id.result)


    }

    /**
     * 功能防抖
     */
    private fun setOnClickNoRepeat() {
        /*
         * 1. 此处采用了RxBinding：RxView.clicks(button) = 对控件点击进行监听，需要引入依赖：compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
         * 2. 传入Button控件，点击时，都会发送数据事件（但由于使用了throttleFirst（）操作符，所以只会发送该段时间内的第1次点击事件）
         **/
        RxView.clicks(submit).throttleFirst(2, TimeUnit.SECONDS)  // 才发送 2s内第1次点击按钮的事件
            .subscribe(object : Observer<Any> {
                override fun onSubscribe(d: Disposable) {
                    Log.e(TAG, "onSubscribe")
                }

                override fun onNext(t: Any) {
                    Log.e(TAG, "onNext$t")
                }

                override fun onError(e: Throwable) {
                    Log.e(TAG, "onError")
                }

                override fun onComplete() {
                    Log.e(TAG, "onComplete")
                }
            })
    }

    /**
     * 联合判断
     */
    private fun joinEvaluation() {
        /*
         * 步骤2：为每个EditText设置被观察者，用于发送监听事件
         * 说明：
         * 1. 此处采用了RxBinding：RxTextView.textChanges(name) = 对对控件数据变更进行监听（功能类似TextWatcher），需要引入依赖：compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
         * 2. 传入EditText控件，点击任1个EditText撰写时，都会发送数据事件 = Function3（）的返回值（下面会详细说明）
         * 3. 采用skip(1)原因：跳过 一开始EditText无任何输入时的空值
         **/
        val nameObservable = RxTextView.textChanges(name).skip(1)
        val ageObservable = RxTextView.textChanges(age).skip(1)
        val jobObservable = RxTextView.textChanges(job).skip(1)

        /*
         * 步骤3：通过combineLatest（）合并事件 & 联合判断
         **/
        val subscribe = Observable.combineLatest(nameObservable,
            ageObservable,
            jobObservable,
            object : Function3<CharSequence, CharSequence, CharSequence, Boolean> {
                override fun invoke(p1: CharSequence, p2: CharSequence, p3: CharSequence): Boolean {
                    /*
                     * 步骤4：规定表单信息输入不能为空
                     **/
                    // 1. 姓名信息
                    val isUserNameValid = !TextUtils.isEmpty(name.text)
                    // 除了设置为空，也可设置长度限制
                    // boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) && (name.getText().toString().length() > 2 && name.getText().toString().length() < 9);

                    // 2. 年龄信息
                    val isUserAgeValid = !TextUtils.isEmpty(age.text)
                    // 3. 职业信息
                    val isUserJobValid = !TextUtils.isEmpty(job.text)

                    /*
                     * 步骤5：返回信息 = 联合判断，即3个信息同时已填写，"提交按钮"才可点击
                     **/
                    return isUserNameValid && isUserAgeValid && isUserJobValid
                }
            }).subscribe(object : Consumer<Boolean> {
            override fun accept(b: Boolean) {
                /*
                 * 步骤6：返回结果 & 设置按钮可点击样式
                 **/
                submit.isEnabled = b;
            }
        }, object : Consumer<Throwable> {
            override fun accept(t: Throwable?) {

            }
        })
    }

    /**
     * 合并数据源
     */
    private fun dataSourceMerge() {
        var result = "数据源来自 = "
        /*
         * 设置第1个Observable：通过网络获取数据
         * 此处仅作网络请求的模拟
         **/
        val network = Observable.just("网络")

        /*
         * 设置第2个Observable：通过本地文件获取数据
         * 此处仅作本地文件请求的模拟
         **/
        val file = Observable.just("本地文件")

        Observable.merge(network, file).subscribe(object : Observer<String> {
            override fun onSubscribe(d: Disposable) {
                Log.e(TAG, "onSubscribe")
            }

            override fun onNext(s: String) {
                Log.e(TAG, "onNext:$s")
                result += "$s+"
            }

            override fun onError(e: Throwable) {
                Log.e(TAG, "onError")
            }

            override fun onComplete() {
                Log.e(TAG, "onComplete:$result")
            }

        })
    }

    /**
     *
     */
    private fun getDataFromCache() {
        val memoryCache: String? = null
        val diskCache = "从磁盘中获取数据"
        /*
         * 设置第1个Observable：检查内存缓存是否有该数据的缓存
         **/
        val menory = Observable.create(object : ObservableOnSubscribe<String> {
            override fun subscribe(emitter: ObservableEmitter<String>) {
                // 先判断内存缓存有无数据
                if (memoryCache != null) {
                    // 若有该数据，则发送
                    emitter.onNext(memoryCache)
                } else {
                    // 若无该数据，则直接发送结束事件
                    emitter.onComplete();
                }
            }
        })
        /*
         * 设置第2个Observable：检查磁盘缓存是否有该数据的缓存
         **/
        val disk = Observable.create(object : ObservableOnSubscribe<String> {
            override fun subscribe(emitter: ObservableEmitter<String>) {
                // 先判断内存缓存有无数据
                if (diskCache != null) {
                    // 若有该数据，则发送
                    emitter.onNext(diskCache)
                } else {
                    // 若无该数据，则直接发送结束事件
                    emitter.onComplete()
                }
            }
        })
        /*
        * 设置第3个Observable：通过网络获取数据
        **/
        val network = Observable.just("从网络中获取数据")
        /*
         * 通过concat（） 和 firstElement（）操作符实现缓存功能
         **/

        // 1. 通过concat（）合并memory、disk、network 3个被观察者的事件（即检查内存缓存、磁盘缓存 & 发送网络请求）
        //    并将它们按顺序串联成队列
        Observable.concat(menory, disk, network)
            // 2. 通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件），即依次判断检查memory、disk、network
            .firstElement()
            // 即本例的逻辑为：
            // a. firstElement()取出第1个事件 = memory，即先判断内存缓存中有无数据缓存；由于memoryCache = null，即内存缓存中无数据，所以发送结束事件（视为无效事件）
            // b. firstElement()继续取出第2个事件 = disk，即判断磁盘缓存中有无数据缓存：由于diskCache ≠ null，即磁盘缓存中有数据，所以发送Next事件（有效事件）
            // c. 即firstElement()已发出第1个有效事件（disk事件），所以停止判断。
            .subscribe(object : MaybeObserver<String> {
                override fun onSubscribe(d: Disposable) {
                }

                override fun onSuccess(s: String) {
                    Log.e(TAG, "最终获取的数据源 = $s")
                }

                override fun onError(e: Throwable) {
                }

                override fun onComplete() {
                }
            })
    }

    /**
     * 联想搜索
     */
    private fun search() {
        /*
        * 说明
        * 1. 此处采用了RxBinding：RxTextView.textChanges(name) = 对对控件数据变更进行监听（功能类似TextWatcher），需要引入依赖：compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
        * 2. 传入EditText控件，输入字符时都会发送数据事件（此处不会马上发送，因为使用了debounce（））
        * 3. 采用skip(1)原因：跳过 第1次请求 = 初始输入框的空字符状态
        **/
        RxTextView.textChanges(search).skip(1).debounce(1, TimeUnit.SECONDS)
            .subscribe(object : Observer<CharSequence> {
                override fun onSubscribe(d: Disposable) {
                }

                override fun onNext(t: CharSequence) {
                    result.text = "发送给服务器的字符 = $t"
                }

                override fun onError(e: Throwable) {
                }

                override fun onComplete() {
                }

            })
    }

    /**
     * 网络请求出错重连
     */
    private fun networkRequestErrorReconnection() {
        // 可重试次数
        val maxConnectionCount = 3
        // 当前已重试次数
        var currentRetryCount = 0
        // 重试等待时间
        var waitRetryTime = 0L

        val observable = Observable.create(object : ObservableOnSubscribe<String> {
            override fun subscribe(emitter: ObservableEmitter<String>) {
                emitter.onError(IOException("io"))
//                emitter.onNext("网络请求")
            }

        })
        // 发送网络请求 & 通过retryWhen（）进行重试
        // 注：主要异常才会回调retryWhen（）进行重试
        observable.retryWhen(object : Function<Observable<Throwable>, ObservableSource<*>> {
            override fun apply(throwableObservable: Observable<Throwable>): ObservableSource<*> {
                // 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
                return throwableObservable.flatMap(object :
                    Function<Throwable, ObservableSource<*>> {
                    override fun apply(throwable: Throwable): ObservableSource<*> {
                        // 输出异常信息
                        Log.e(TAG, "发生异常 = $throwable")
                        /**
                         * 需求1：根据异常类型选择是否重试
                         * 即，当发生的异常 = 网络异常 = IO异常 才选择重试
                         */
                        if (throwable is IOException) {
                            Log.e(TAG, "属于IO异常，需重试")

                            /**
                             * 需求2：限制重试次数
                             * 即，当已重试次数 < 设置的重试次数，才选择重试
                             */
                            if (currentRetryCount < maxConnectionCount) {
                                currentRetryCount++
                                /**
                                 * 需求2：实现重试
                                 * 通过返回的Observable发送的事件 = Next事件，从而使得retryWhen（）重订阅，最终实现重试功能
                                 *
                                 * 需求3：延迟1段时间再重试
                                 * 采用delay操作符 = 延迟一段时间发送，以实现重试间隔设置
                                 *
                                 * 需求4：遇到的异常越多，时间越长
                                 * 在delay操作符的等待时间内设置 = 每重试1次，增多延迟重试时间1s
                                 */
                                // 设置等待时间
                                waitRetryTime = 1000L + currentRetryCount * 1000
                                Log.e(TAG, "等待时间 =$waitRetryTime");
                                return Observable.timer(waitRetryTime, TimeUnit.MILLISECONDS)
                            } else {
                                // 若重试次数已 > 设置重试次数，则不重试
                                // 通过发送error来停止重试（可在观察者的onError（）中获取信息）
                                return Observable.error<Any>(Throwable("重试次数已超过设置次数 = $currentRetryCount ，即 不再重试"))
                            }
                        } else {
                            // 若发生的异常不属于I/O异常，则不重试
                            // 通过返回的Observable发送的事件 = Error事件 实现（可在观察者的onError（）中获取信息）
                            return Observable.error<Any>(Throwable("发生了非网络异常（非I/O异常）"))
                        }
                    }
                })
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(object : Observer<String> {
                override fun onSubscribe(d: Disposable) {

                }

                override fun onNext(s: String) {
                    // 接收服务器返回的数据
                    Log.e(TAG, "发送成功$s")
                }

                override fun onError(e: Throwable) {
                    // 获取停止重试的信息
                    Log.e(TAG, e.toString())
                }

                override fun onComplete() {
                }

            })
    }

    /**
     * 网络请求轮询
     */
    private fun networkRequestPolling() {
        var count = 1
        // 设置变量 = 模拟轮询服务器次数
        val repeatCount = 3
        val observable = Observable.just("网络请求")
        // 发送网络请求 & 通过repeatWhen（）进行轮询
        observable.repeatWhen(object : Function<Observable<Any>, ObservableSource<*>> {
            // 在Function函数中，必须对输入的 Observable<Object>进行处理，此处使用flatMap操作符接收上游的数据
            override fun apply(objectObservable: Observable<Any>): ObservableSource<*> {
                // 将原始 Observable 停止发送事件的标识（Complete（） /  Error（））转换成1个 Object 类型数据传递给1个新被观察者（Observable）
                // 以此决定是否重新订阅 & 发送原来的 Observable，即轮询
                // 此处有2种情况：
                // 1. 若返回1个Complete（） /  Error（）事件，则不重新订阅 & 发送原来的 Observable，即轮询结束
                // 2. 若返回其余事件，则重新订阅 & 发送原来的 Observable，即继续轮询
                return objectObservable.flatMap(object : Function<Any, ObservableSource<*>> {
                    override fun apply(t: Any): ObservableSource<*> {
                        // 加入判断条件：当轮询次数 = 3次后，就停止轮询
                        if (count < repeatCount) {
                            // 若轮询次数＜3次，则发送Next事件以继续轮询
                            // 注：此处加入了timer操作符，作用 = 延迟一段时间发送（此处设置 = 2s），以实现轮询间间隔设置
                            return Observable.timer(2, TimeUnit.SECONDS)
                        } else {
                            // 此处选择发送onError事件以结束轮询，因为可触发下游观察者的onError（）方法回调
                            return Observable.error<Any>(Throwable("轮询结束"))
                        }
                    }
                })
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(object : Observer<String> {
                override fun onSubscribe(d: Disposable) {
                }

                override fun onNext(s: String) {
                    // 结收服务器返回的数据
                    Log.e(TAG, "发送成功:$s")
                    count++
                }

                override fun onError(e: Throwable) {
                    // 获取轮询结束信息
                    Log.e(TAG, e.toString())
                }

                override fun onComplete() {
                }
            })
    }
}